package com.itzhanghada.rebitmq.mq;

import com.itzhanghada.rebitmq.enums.TransactionStatus;
import com.itzhanghada.rebitmq.redis.RedisStorageService;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

@Component
public class MessageConsumer {
    private static final Logger logger = LoggerFactory.getLogger(MessageConsumer.class);
    
    @Autowired
    private RedisStorageService storageService;
    
    @RabbitListener(queues = "${rebit.mq.queue.name:rebit.queue}")
    public void handleMessage(Message message, Channel channel) throws IOException {
        String txId = (String) message.getMessageProperties().getHeaders().get("TX_ID");
        String serviceId = "order-service"; // 根据实际服务标识
        
        if (txId == null) {
            logger.warn("Received message without TX_ID, ignoring");
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }
        
        // 检查幂等性
        if (!storageService.markMessageConsumed(txId, serviceId)) {
            logger.info("Message already processed, ignoring, txId: {}", txId);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }
        
        try {
            // 处理消息
            processMessage(message.getBody());
            
            // 更新事务状态为CONSUMED
            storageService.updateTransactionStatus(txId, TransactionStatus.CONSUMED);
            
            // 确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            
            logger.info("Message processed successfully, txId: {}", txId);
        } catch (Exception e) {
            logger.error("Failed to process message, txId: {}", txId, e);
            
            // 处理失败，拒绝消息（可以配置重试或进入死信队列）
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }
    
    private void processMessage(byte[] body) {
        // 实现具体的消息处理逻辑
        try {
            String messageBody = new String(body, StandardCharsets.UTF_8);
            logger.info("Processing message: {}", messageBody);
            
            // 这里根据实际业务需求解析和处理消息
            // ObjectMapper mapper = new ObjectMapper();
            // MyBusinessObject obj = mapper.readValue(messageBody, MyBusinessObject.class);
            // businessService.process(obj);
        } catch (Exception e) {
            throw new RuntimeException("Failed to process message", e);
        }
    }
}